package com.atbeijing.bigdata.spark.core.rdd.operator.action

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark06_Oper_Action {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setMaster("local[*]").setAppName("ActionOperator")
        val sc = new SparkContext(conf)

        // TODO 算子 - 行动
        val rdd = sc.makeRDD(
            List(
                ("a", 1), ("a", 2),("a", 3),
                ("b", 4), ("b", 5),("b", 6)
            )
        )

        //期望得到 Map【 (a, 6), (b, 15) 】

        // reduce行动算子
        // KV => Map
        // ("a", 1) => Map(("a", 1))
        // ("a", 2) => Map(("a", 2))
        // ("a", 3) => Map(("a", 3))
        // ("b", 4) => Map(("b", 4))
        // Map(("a", 1)), Map(("a", 2))
        // Map(("a", 3)),Map(("a", 3))
        // Map(("a", 6)),Map(("b", 4))
        // Map(("a", 6), ("b", 4))

        val r1: RDD[Map[String, Int]] = rdd.map(kv => Map(kv))
        val stringToInt: Map[String, Int] = r1.reduce((map1, map2) => {
            // scala 两个map的合并
            map1.foldLeft(map2)({
                  //就是map1 ,map2
                (map, kv) => {
                    val k = kv._1
                    val v = kv._2
                    val oldvalue = map.getOrElse(k, 0)
                    map.updated(k, v + oldvalue)
                }
            })
        })

        println(stringToInt)


        sc.stop()

    }
}
